package top.bulk.delay.handler;

import cn.hutool.json.JSONUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import top.bulk.delay.entity.OrderEntity;
import top.bulk.delay.mapper.OrderMapper;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 基于 redis zset 实现 ,当然，实际生产建议使用 redisson
 *
 * @author 散装java
 * @date 2024-08-13
 */
@Component
@Slf4j
public class RedisZsetDelayHandler implements DelayHandler<OrderEntity> {
    private static final String KEY = "delaySet:orderDelaySet";
    @Resource
    StringRedisTemplate stringRedisTemplate;
    @Resource
    private OrderMapper orderMapper;

    @Override
    public void handle(OrderEntity orderEntity, Integer second) {
        long currentTimeMillis = System.currentTimeMillis();
        OrderCache cache = new OrderCache(orderEntity.getId(), second);
        stringRedisTemplate.opsForZSet().add(KEY, JSONUtil.toJsonStr(cache), currentTimeMillis + second * 1000);
    }

    @PostConstruct
    public void init() {
        Thread pollingThread = new Thread(() -> {
            while (true) {
                pollingRedis();
                // 100ms一次，可根据业务调整时间，redis 可以承载约 10w qps
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        pollingThread.setDaemon(true);
        pollingThread.start();
    }

    /**
     * 轮询 redis
     */
    public void pollingRedis() {
        int limit = 100;
        // 限制数量，防止炸内存，依据实际场景处理
        Set<String> set = stringRedisTemplate.opsForZSet().rangeByScore(KEY, 0, System.currentTimeMillis(), 0, limit);
        if (set != null && set.size() > 0) {
            set.forEach((obj) -> {
                // 先去移除，防止重复消费，移除成功，才可以处理
                Long remove = stringRedisTemplate.opsForZSet().remove(KEY, obj);
                if (remove != null && remove > 0) {
                    // 实际生产，注意消费速度，可以开多线程处理
                    OrderCache cache = JSONUtil.toBean(obj, OrderCache.class);
                    OrderEntity orderEntity = orderMapper.queryById(cache.getId());
                    if (orderEntity.canClose(cache.getSecond())) {
                        orderMapper.closeOrder(orderEntity.getId());
                        log.info("订单：{} 成功关闭", orderEntity.getOrderNo());
                    }
                }
            });
        }
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class OrderCache {
        private Integer id;
        private Integer second;
    }
}
